package com.shujia.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo10Join {
  def main(args: Array[String]): Unit = {
    // 创建Spark Context
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo10Join")
    conf.setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    // 读取学生数据及分数数据 并将每一行数据转换成样例类对象
    val stuRDD: RDD[Student] = sc
      .textFile("Spark/data/students.txt")
      .map(line => {
        val splits: Array[String] = line.split(",")
        val id: String = splits(0)
        val name: String = splits(1)
        val age: Int = splits(2).toInt
        val gender: String = splits(3)
        val clazz: String = splits(4)
        Student(id, name, age, gender, clazz)
      })

    val scoreRDD: RDD[Score] = sc.textFile("Spark/data/score.txt")
      .map(line => {
        val splits: Array[String] = line.split(",")
        val id: String = splits(0)
        val subId: String = splits(1)
        val score: Int = splits(2).toInt
        Score(id, subId, score)
      })

    /**
     * 将学生及分数数据关联 [id,name,clazz,subId,score]
     * 关联之前需要将RDD变成KV格式 一般将关联的字段作为Key Value自定
     */
    //
    val stuKVRDD: RDD[(String, Student)] = stuRDD
      .map(stu => (stu.id, stu))

    val scoreKVRDD: RDD[(String, Score)] = scoreRDD
      .map(sco => (sco.id, sco))

    val joinRDD: RDD[(String, (Student, Score))] = stuKVRDD.join(scoreKVRDD)

    joinRDD
      .map(kv => {
        val id: String = kv._1
        val stuScoreTuple: (Student, Score) = kv._2
        val stu: Student = stuScoreTuple._1
        val score: Score = stuScoreTuple._2
        s"$id,${stu.name},${stu.clazz},${score.subId},${score.score}"
      }).foreach(println)

    // 使用match 简写
    joinRDD
      .map {
        case (id: String, (stu: Student, score: Score)) =>
          s"$id,${stu.name},${stu.clazz},${score.subId},${score.score}"
      }.foreach(println)

    val leftJoinRDD: RDD[(String, (Student, Option[Score]))] = stuKVRDD
      .leftOuterJoin(scoreKVRDD)

    leftJoinRDD
      .map {
        case (id: String, (stu: Student, Some(score))) =>
          s"$id,${stu.name},${stu.clazz},${score.subId},${score.score}"
        case (id: String, (stu: Student, None)) =>
          s"$id,${stu.name},${stu.clazz},null,null"
      }.foreach(println)



  }

  case class Student(id: String, name: String, age: Int, gender: String, clazz: String)

  case class Score(id: String, subId: String, score: Int)

}
